package org.ala.tree.lsm.sstable.block;

import java.nio.ByteBuffer;

import org.ala.tree.lsm.KV;
import org.ala.tree.lsm.tools.KryoTools;
import org.ala.tree.lsm.tools.compression.PrefixCompression;
import org.ala.tree.lsm.tools.compression.PrefixCompression.Result;

/**
 * 数据块
 * <p>	二进制格式：
 * <p>	[DataBlock头 [条目头|键差key|值], [数据头|键差key|值], ..., ]
 * <p>	DataBlock头：
 * <p>		dataBlock长度（4字节）
 * <p>	条目头：
 * <p>		key差异下标：4字节
 * <p>		key总长：4字节
 * <p>		val总长：4字节
 * <p>		差异key：
 * <p>		val
 *
 * @author ala
 * @date 2025年5月28日
 */
public class DataBlock<V> {
	/**
	 * 	数据块头长度
	 */
	public static final int HEAD_ITEM_LEN = 12;
	/**
	 * 	容量（单位：键值对数目）
	 */
	protected int capacity = 16;
	/**
	 * 	当前追加数量
	 */
	protected int count = 0;
	
	
	/**
	 * 	首key
	 */
	protected String key;
	/**
	 * 	key集合
	 */
	protected String[] keys;
	/**
	 * 	value集合
	 */
	protected Object[] vals;
	/**
	 * 	key前缀压缩
	 */
	protected PrefixCompression pc;
	/**
	 * 	前缀压缩结果
	 */
	protected PrefixCompression.Result[] rs;
	
	
	public DataBlock(int capacity) {
		this.capacity = capacity;
		keys = new String[capacity];
		vals = new Object[capacity];
		rs = new PrefixCompression.Result[capacity];
		pc = new PrefixCompression();
	}
	
	
	public boolean isFull() {return count == capacity;}
	
	
	/**
	 * 	追加键值对
	 */
	public void append(String k, V v) {
		if (count == 0) {key = k;}
		keys[count] = k;
		vals[count] = v;
		rs[count] = pc.compression(k);
		count++;
	}
	/**
	 * 	转为二进制
	 */
	public ByteBuffer encode() {
		//	数据总长
		int c = 0;
		EncodeResult[] ers = new EncodeResult[count];
		for (int i = 0 ; i < count ; i++) {
			ers[i] = EncodeResult.build(
					rs[i].getI(), 
					KryoTools.serializ(rs[i].getStr()), 
					KryoTools.serializ(vals[i]));
			//	差异下标（4字节），key总长（4字节），val总长（4字节）
			c += 12;
			c += ers[i].key.length;
			c += ers[i].val.length;
		}
		//	构建ByteBuffer
		ByteBuffer buf = ByteBuffer.allocateDirect(c + 4);
		buf.putInt(c);
		for (int i = 0 ; i < count ; i++) {
			buf.putInt(ers[i].i);
			buf.putInt(ers[i].key.length);
			buf.putInt(ers[i].val.length);
			buf.put(ers[i].key);
			buf.put(ers[i].val);
		}
		buf.flip();
		return buf;
	}
	/**
	 * 	解码
	 * 	<p>	只解码出keys和vals
	 */
	public static <V> DataBlock<V> decode(int capacity, ByteBuffer buf, Class<V> vclass) {
		DataBlock<V> db = new DataBlock<>(capacity);
		while (buf.remaining() > 0) {
			//	key差异下标，key总长，val总长
			int pi = buf.getInt(), kl = buf.getInt(), vl = buf.getInt();
			//	差异key，转换为真实key
			byte[] keyBytes = new byte[kl];
			buf.get(keyBytes);
			String key = KryoTools.deSerialize(keyBytes, String.class);
			key = db.pc.revert(Result.build(key, pi));
			//	值
			byte[] valBytes = new byte[vl];
			buf.get(valBytes);
			V val = KryoTools.deSerialize(valBytes, vclass);
//			db.pc.revert(Result.build(key, pi));
			
			db.keys[db.count] = key;
			db.vals[db.count] = val;
			db.count++;
		}
		
		return db;
	}
	/**
	 * 	迭代指针
	 */
	protected int pos = 0;
	public void resetIteratprPos() {pos = 0;}
	public boolean hasNext() {
		return pos < count;
	}
	@SuppressWarnings("unchecked")
	public KV<V> next() {
		String key = keys[pos];
		V v = (V) vals[pos];
		pos++;
		return KV.build(key, v);
	}
	@SuppressWarnings("unchecked")
	public KV<V> current() {
		String key = keys[pos];
		V v = (V) vals[pos];
		return KV.build(key, v);
	}
	
	
	public int getCount() {
		return count;
	}
	public void setCount(int count) {
		this.count = count;
	}
	public String getKey() {
		return key;
	}
	public void setKey(String key) {
		this.key = key;
	}
	public String[] getKeys() {
		return keys;
	}
	public void setKeys(String[] keys) {
		this.keys = keys;
	}


	/**
	 * 	编码结果
	 */
	public static class EncodeResult {
		/**
		 * 	key的差异下标
		 */
		int i;
		/**
		 * 	key的编码结果
		 */
		protected byte[] key;
		/**
		 * 	值的编码结果
		 */
		protected byte[] val;
		
		public static EncodeResult build(int i, byte[] key, byte[] val) {
			EncodeResult re = new EncodeResult();
			re.i = i;
			re.key = key;
			re.val = val;
			return re;
		}
		
		public byte[] getKey() {
			return key;
		}
		public void setKey(byte[] key) {
			this.key = key;
		}
		
		public byte[] getVal() {
			return val;
		}
		public void setVal(byte[] val) {
			this.val = val;
		}
		public int getI() {
			return i;
		}
		public void setI(int i) {
			this.i = i;
		}
	}
}
